package com.liu.hadoop.clickStreamETL.mapreduce;

import java.io.IOException;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

/**
 * @Description: MRreducer类，主要负责输出数据，和简单计数。
 * @author 刘章程
 */
public class ClickStreamReducer extends Reducer<Text, Text, NullWritable, Text> {
	
	/**
	 * @Fields SPLIT_CHAR : 字符拆分格式
	 */
	public static final String SPLIT_CHAR = "\t";
	
	/**
	 * @Fields preSessionId : 表示前一个sessionId
	 */
	public static String preSessionId = "-";//66：改进版把他变为静态变量
	/**
	 * @Fields csvp : 改进版把csvp变为静态变量，并移到成员函数外面
	 */
	static int csvp = 0;
	
	/* (non-Javadoc)
	 * @see org.apache.hadoop.mapreduce.Reducer#reduce(KEYIN, java.lang.Iterable, org.apache.hadoop.mapreduce.Reducer.Context)
	 */
	@Override
	protected void reduce(Text key, Iterable<Text> values, Context context)
			throws IOException, InterruptedException {
		
		String sessionId = key.toString().split("&")[0];
		
		if (preSessionId.equals("-")) {
			csvp = 1;
			preSessionId = sessionId;
		} else {
			// 判断相同则 累加csvp
			if (preSessionId.equals(sessionId)) {
				csvp ++;
			} else {
			// 如果不同，说明是新的session,重置preSessionId 和 csvp
				csvp = 1;
				preSessionId = sessionId;
			}
		}
		
		// 按照clickstream_log的格式再末尾加上csvp
		String reduceOutValue = values.iterator().next().toString() + SPLIT_CHAR + csvp;
		
		context.write(NullWritable.get(), new Text(reduceOutValue));
	}

}
